/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package br.ufba.lasid.ds.fd;

import br.ufba.lasid.base.ResourceFactory;
import br.ufba.lasid.base.api.IResource;
import br.ufba.lasid.base.api.IResource.Kinds;
import br.ufba.lasid.base.api.IResource.States;
import br.ufba.lasid.base.api.IResourceFactory;
import br.ufba.lasid.base.api.IResourceList;
import br.ufba.lasid.base.api.IResourcetable;
import br.ufba.lasid.comm.Hosttable;
import br.ufba.lasid.comm.api.ICommunicator;
import br.ufba.lasid.comm.api.IHost;
import br.ufba.lasid.ds.fd.api.IFailureDetector;
import br.ufba.lasid.ds.fd.api.IFailureDetector.MessageKinds;
import br.ufba.lasid.ds.fd.api.IFailureDetector.ModuleTypes;
import br.ufba.lasid.ds.fd.api.IFailureDetectorMessageHandler;
import br.ufba.lasid.utils.Agenda;
import br.ufba.lasid.utils.Schedule;
import java.net.ConnectException;
import java.net.NoRouteToHostException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 *
 * @author aliriosa
 */
public class FailureDetectorMessageHandler implements IFailureDetectorMessageHandler{
    
    IFailureDetector detector;
    private static final Log logger = LogFactory.getLog(FailureDetectorMessageHandler.class);

    public FailureDetectorMessageHandler() {
    }

    public FailureDetectorMessageHandler(IFailureDetector detector) {
        this.detector = detector;
    }
        

    @Override
    public void handle(String hostID, String message) throws Exception {
        
        //logger.info(message + " from " + hostID + " has been received.");
        
        MessageKinds msgKind = IFailureDetector.Utils.getMessageType(message);      
        
//        logger.info(" message kind " + msgKind);

        switch(msgKind){
            case StateReport:
                handleReporting(hostID, message);
                break;
            case CheckState:
                handleChecking (hostID, message);
        }
        
    }
    
    public void handleReporting(String hostID, String message) throws Exception{
        
        IHost localhost = detector.getLocalhost();
        
        if(!localhost.getID().equals(hostID)){
            
            Hosttable hosttable = detector.getHosttable();
            IHost host = hosttable.get(hostID);            
            
            if(host == null){
                IResourceFactory hfactory = 
                        ResourceFactory.getResourceFactory(Kinds.Host);
                
                host = hfactory.create(hostID);                                
                hosttable.put(hostID, host);                
            }
            
            final Agenda agenda = detector.getAgenda();            
            final Schedule schedule;
            
            synchronized(agenda){
                if(agenda.containsKey(hostID)){
                    schedule = agenda.get(hostID);
                }else{
                    schedule = new Schedule();
                    agenda.put(hostID, schedule);
                    agenda.notify();
                }
            }
            
            synchronized(schedule){
                schedule.clear();
                schedule.notify();
            }
            
            host.setState(States.Up);
            
            IResourceList resources = IFailureDetector.Utils.getResourceListFromMessage(message);
            
            for(IResource resource : resources){
                IResourcetable resourcetable = host.getResourcetable();
                IResource hresource = resourcetable.get(resource.getID());
                
                if(hresource == null){
                    hresource = resource;
                }
                
                hresource.setState(resource.getState());
                hresource.setParent(host);
                
                resourcetable.add(hresource);
            }
            //IFailureDetector.Utils.getHostIDFromResourceAddress(hostID);
        }        
    }
    
    public void handleChecking(String hostID, String request) throws Exception{
        
        IHost localhost = detector.getLocalhost();
        
        if(localhost.getID().equals(hostID)){
            return;
        }
        
        ICommunicator comm = detector.getCommunicator();               
        
        /**
         * We use the message as a opportunity to update the host state.
         */
        Hosttable hosttable = detector.getHosttable();
        IHost host = hosttable.get(hostID);            

        if(host != null){
            host.setState(States.Up);
        }
        
        /**
         * Here, we'll send a response for the check state message.
         */                
        String response = IFailureDetector.Utils.newMessage(
                MessageKinds.StateReport, localhost.getResourcetable()
        );
        
        /**
         * Instantiate a new thread at each sending instant can bring some 
         * performance issues -- we must review it later.
         */
        new Sender(host, response).start();
        //comm.send(hostID, detector.getName(), response);
        
        
        
    }
    

    @Override
    public ModuleTypes getModuleType() throws Exception {
        return ModuleTypes.MessageHandler;
    }

    @Override
    public void setFailureDetector(IFailureDetector detector) throws Exception {
        this.detector = detector;
    }

    @Override
    public void startup() throws Exception {
        new Thread(this).start();
    }

    @Override
    public void run() {
        //do nothing
    }

    @Override
    public String getName() throws Exception {
        return getModuleType().name();
    }

    @Override
    public void receive(String host, String component, String message) throws Exception {
        if(component != null && component.equals(detector.getName())){
            handle(host, message);
        }
    }
    
    class Sender extends Thread{
        IHost host;
        String message;
        
        public Sender() {
        }

        public Sender(IHost host, String message) {
            this.host = host;
            this.message = message;
        }
                
        @Override
        public void run() {
            String hostID = host.getID();                    
            try{
                detector.getCommunicator().send(hostID, detector.getName(), message);  
            }catch(ConnectException ce){
                host.setState(IResource.States.Down);                        
                //ce.printStackTrace();
            }catch(NoRouteToHostException nre){
                host.setState(IResource.States.Down);
            }catch(Exception e){
                e.printStackTrace();
            }
        }
        
    }
    
    
}
